package com.vhall.component.service.exam.tasks;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.time.Duration;

/**
 * @author yongzheng.xu
 * @date 2021/5/24 11:48
 */
@Slf4j
@Configuration
public class ExamMsgStreamConfig {

    @Autowired
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    @Autowired
    private ExamMsgStreamListener listener;

    @Value("${vhall.exam.stream-name:exam-stream}")
    public String streamName;

    @Value("${vhall.exam.group-name:exam-group}")
    private String groupName;

    @Value("${vhall.exam.consumer-name:exam-consumer}")
    private String consumerName;

    @Bean(name = "Exam-StreamMessageListener")
    public StreamMessageListenerContainer<String, ObjectRecord<String, String>> getStreamMessageListenerContainer(RedisConnectionFactory factory) {

        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, String>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                        .builder()
                        .pollTimeout(Duration.ofSeconds(1))
                        .batchSize(10)
                        .targetType(String.class)
                        .executor(threadPoolTaskExecutor)
                        .build();

        StreamMessageListenerContainer<String, ObjectRecord<String, String>> container
                = StreamMessageListenerContainer.create(factory, options);

        //指定消费最新的消息
        StreamOffset<String> offset = StreamOffset.create(streamName, ReadOffset.lastConsumed());

        //创建消费者
        Consumer consumer = Consumer.from(groupName, consumerName);

        StreamMessageListenerContainer.StreamReadRequest<String> streamReadRequest =
                StreamMessageListenerContainer.StreamReadRequest.builder(offset)
                        .errorHandler((error) -> log.error(error.getMessage()))
                        .cancelOnError(e -> false)
                        .consumer(consumer)
                        //关闭自动ack确认
                        .autoAcknowledge(false)
                        .build();
        //指定消费者对象
        container.register(streamReadRequest, listener);
        container.start();
        return container;
    }

}
